package org.action.order.util;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.lettuce.core.RedisBusyException;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.*;

import static org.action.order.util.StaticVariable.Action_Order_Stream_Group;
import static org.action.order.util.StaticVariable.Action_Order_Stream_Queue;
@Slf4j
@Component
public class StreamRedisList {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    private static final ObjectMapper objectMapper=new ObjectMapper(); // Spring Boot 自动配置的 Jackson 序列化器

    // 添加连接工厂引用
    @Autowired
    private RedisConnectionFactory redisConnectionFactory;





    //向队列中添加数据
    public void addToMessageQueue(Object message) throws JsonProcessingException {
        // 1.将数据转为JSON
        String json = objectMapper.writeValueAsString(message);
        Map<String, String> map = new HashMap<>();  // 创建一个 Map 用来存储消息内容
        map.put("actionOrder", json);  // 将消息内容添加到 Map 中，键是 "content"，值是传入的内容

        // 在消息队列中添加消息，调用 StringRedisTemplate 的 opsForStream 方法
        RecordId recordId = stringRedisTemplate.opsForStream()  // 获取操作 Redis Stream 的操作对象
                        .add(StreamRecords.objectBacked(map)  // 创建一个 Stream 记录，将 Map 转化为对象记录
                        .withStreamKey(Action_Order_Stream_Queue));  // 设置该记录属于的 Stream（消息队列）的名称
        // 输出记录的 ID，表示消息已经成功发送
        System.out.println("消息发送成功，id: " + recordId.getValue());  // 打印消息的 ID，表明该消息已经被成功加入到 Stream 中
    }

    //从队列中获取消息
    public <T> List<Map<String, T>> getFromMessageQueue(long timeout,String consumer, Class<T> clazz) throws JsonProcessingException {
        // 设置 Stream 读取的阻塞超时，
        StreamReadOptions options = StreamReadOptions.empty().block(Duration.ofSeconds(timeout)).count(10);

        // 从指定的消费者组中读取消息，">" 表示只消费未被消费过的消息
        List<MapRecord<String, Object, Object>> messages = stringRedisTemplate.opsForStream().read(
                Consumer.from(Action_Order_Stream_Group, consumer),  // 指定消费者组和消费者名称
                options,  // 设置读取选项，包含阻塞时间
                StreamOffset.create(Action_Order_Stream_Queue, ReadOffset.lastConsumed())  // 从最后消费的消息开始读取
        );
        if (messages == null || messages.isEmpty()) {
            return Collections.emptyList();
        }
        //创建一个列表，用于存储读取到的的消息，
        List<Map<String, T>> result = new ArrayList<>();
        //对读取到的消息进行处理
        for(MapRecord<String, Object, Object> message : messages){
            Map<String, T> messageMap = new HashMap<>();
            //获取消息的ID
            String messageId = String.valueOf(message.getId());  // 获取消息的唯一标识符（ID）
            //获取消息的值
            Map<Object, Object> value = message.getValue();
            //将消息的值转换为JSON字符串
            String json = (String) value.get("actionOrder");
            System.out.println("消息id"+messageId);
            //将JSON字符串转换为对象
            T actionOrder = objectMapper.readValue(json, clazz);
            //将对象添加到Map中，以消息的ID作为键
            messageMap.put(messageId, actionOrder);
            result.add(messageMap);
        }
        return result;
    }

    //处理pedding-list里的消息
    public <T> List<Map<String, T>> processPendingMessages(String consumer, Class<T> clazz) throws JsonProcessingException {
        // 检查连接工厂状态
        if (redisConnectionFactory instanceof LettuceConnectionFactory) {
            LettuceConnectionFactory lettuceFactory = (LettuceConnectionFactory) redisConnectionFactory;
            if (!lettuceFactory.isRunning()) {
                // 尝试自动重启连接
                try {
                    lettuceFactory.start();
                    log.warn("Redis连接工厂已停止，已尝试重新启动");
                } catch (Exception e) {
                    log.error("Redis连接工厂重启失败", e);
                    return Collections.emptyList();
                }
            }
        }
        // 设置 Stream 读取的阻塞超时，
        StreamReadOptions options = StreamReadOptions.empty().count(10);

        // 从指定的消费者组中读取消息，">" 表示只消费未被消费过的消息
        List<MapRecord<String, Object, Object>> messages = stringRedisTemplate.opsForStream().read(
                Consumer.from(Action_Order_Stream_Group, consumer),  // 指定消费者组和消费者名称
                options,  // 设置读取选项
                StreamOffset.create(Action_Order_Stream_Queue, ReadOffset.from("0"))  // 从最后消费的消息开始读取
        );
        if (messages == null || messages.isEmpty()) {
            return Collections.emptyList();
        }
        //创建一个列表，用于存储读取到的的消息，
        List<Map<String, T>> result = new ArrayList<>();
        //对读取到的消息进行处理
        for(MapRecord<String, Object, Object> message : messages){
            Map<String, T> messageMap = new HashMap<>();
            //获取消息的ID
            String messageId = String.valueOf(message.getId());  // 获取消息的唯一标识符（ID）
            //获取消息的值
            Map<Object, Object> value = message.getValue();
            //将消息的值转换为JSON字符串
            String json = (String) value.get("actionOrder");
            //将JSON字符串转换为对象
            T actionOrder = objectMapper.readValue(json, clazz);
            System.out.println("处理消息:"+actionOrder);
            //将对象添加到Map中，以消息的ID作为键
            messageMap.put(messageId, actionOrder);
            result.add(messageMap);
        }
        return result;
    }

    //确认消息
    public void confirmMessage(String messageId) {
        stringRedisTemplate.opsForStream().acknowledge(
                Action_Order_Stream_Queue,
                Action_Order_Stream_Group,
                messageId
        );
    }
    // ===== 新增方法 =====
    public boolean isRedisAvailable() {
        try {
            if (redisConnectionFactory instanceof LettuceConnectionFactory) {
                LettuceConnectionFactory factory = (LettuceConnectionFactory) redisConnectionFactory;
                return factory.isRunning() && factory.getConnection().ping().equals("PONG");
            }
            return stringRedisTemplate.hasKey("__health_check__");
        } catch (Exception e) {
            return false;
        }
    }

}
